use std::{
    collections::VecDeque,
    sync::{atomic::Ordering, Arc},
    time::Duration,
};

use bytes::{BufMut, Bytes, BytesMut};
use http::{Method, Uri, Version};
use log::debug;
use tokio::{
    io::{AsyncReadExt, AsyncWriteExt},
    select,
    sync::oneshot,
    time::timeout,
};

use crate::{
    config::http::{
        core_conf::{HttpCoreRuntimeConf, HEADER_BUF_MIN_WRITEABLE_SIZE},
        main::get_http_main_runtime_conf,
        site_mgr::HttpSiteRuntimeConfManager,
    },
    http::{
        base::{
            conn::{BoxHttpConn, HttpAddressSiteRuntimeConfInfo},
            header_name::HeaderName,
            header_value::HeaderValue,
            headers::{HttpHeader, HttpHeaders},
            parse::{http_read_crlf_line, parse_http_header, HttpResult},
            status_code::{
                get_status_reason, http_body_allow_status, StatusCode,
            },
            Request, Response,
        },
        pool::server_pool::{ServerConnCancelToken, ServerConnPool},
        v1::body::BodyWriter,
    },
    runtime::PROCESS_EXIT,
    utils::{fixed_buffer::FixedBuffer, time::get_now_msec},
};

use crate::error::*;

use super::body::BodyReader;

pub struct ServerSession {
    hc: BoxHttpConn,

    // config
    one_header_buf_size: usize,
    max_header_size: usize,
    read_header_timeout_ms: usize,
    write_header_timeout_ms: usize,
    body_read_timeout_ms: usize,
    body_write_timeout_ms: usize,
    body_buf_size: usize,
    ignore_invalid_header: bool,
    max_discard_request_body: usize,
    server_keepalive_timeout_ms: usize,
    send_100_continue: bool,

    header_bytes: usize,

    // 用于存放待解析数据
    last_buf: Option<FixedBuffer>,

    // request
    req: Request,
    invalid_headers: usize,
    parse_rl_done: bool,
    parse_rh_done: bool,
    want_close: bool,
    content_length: Option<usize>,
    chunked_transfer: bool,
    expect_100_continue: bool,
    // 期望有多少个trailer header.
    expect_trailers: usize,
    body_reader: Option<BodyReader>,

    // response
    resp_status_code: u16,
    resp_no_body: bool,
    pub header_send: bool,
    resp_content_length: Option<usize>,
    resp_chunked_transfer: bool,
    expect_write_trailers: usize,
    body_writer: Option<BodyWriter>,
    write_trailers_done: bool,
    finalized: bool,
}

#[allow(dead_code)]
impl ServerSession {
    pub fn new(hc: BoxHttpConn) -> Self {
        ServerSession {
            hc,
            one_header_buf_size: 0,
            max_header_size: 0,
            read_header_timeout_ms: 0,
            write_header_timeout_ms: 0,
            body_read_timeout_ms: 0,
            body_write_timeout_ms: 0,
            body_buf_size: 0,
            ignore_invalid_header: false,
            max_discard_request_body: 0,
            server_keepalive_timeout_ms: 0,
            send_100_continue: false,

            header_bytes: 0,
            last_buf: None,

            req: Request::new(),
            invalid_headers: 0,
            parse_rl_done: false,
            parse_rh_done: false,
            want_close: false,
            content_length: None,
            chunked_transfer: false,
            expect_100_continue: false,
            expect_trailers: 0,
            body_reader: None,

            resp_status_code: 0,
            resp_no_body: false,
            header_send: false,
            resp_content_length: None,
            resp_chunked_transfer: false,
            expect_write_trailers: 0,
            body_writer: None,
            write_trailers_done: false,
            finalized: false,
        }
    }

    pub fn get_hc(&mut self) -> &mut BoxHttpConn {
        &mut self.hc
    }

    // 该方法有三种返回值.
    // - Ok(Some(req)): 成功读取到header.
    // - Ok(None): 在连接断开、超时等场景中未能读取到任何数据. 此时建议proxy直接关闭连接即可.
    // - Err(err): 读取数据或解析数据发生错误, 此时proxy应该响应4xx给客户端.
    pub async fn read_request(&mut self) -> Result<Option<&mut Request>> {
        // 这是上个请求的conf, 我们先将它设置为invalid
        match &mut self.hc.address_site_conf_info {
            Some(info) => {
                info.valid = false;

                let core_runtime_conf = &info.conf.core_runtime_conf.clone();
                self.update_session_conf(core_runtime_conf);
            }

            None => {
                let main_conf = get_http_main_runtime_conf();
                self.update_session_conf(&main_conf.core_runtime_conf);
            }
        }

        self.hc.requests += 1;

        self.last_buf = Some(FixedBuffer::new(self.one_header_buf_size));

        // 未读取到任何数据时, proxy只需要直接关闭连接即可.
        // 当有数据到达时, proxy通常需要做出response.
        let mut has_data = self.hc.has_pre_data();

        // for http keepalive
        if self.hc.requests > 1 && !has_data {
            if !self.wait_keepalive_request().await? {
                return Ok(None);
            }

            has_data = true;
        }

        let mut next_read_timeout = false;
        let mut read_header_timeout =
            Duration::from_millis(self.read_header_timeout_ms as u64);

        loop {
            let res = self.parse_request()?;
            // parse request header success.
            if res == HttpResult::Complete {
                self.process_request().await?;
                return Ok(Some(&mut self.req));
            }

            if self.header_bytes > self.max_header_size {
                return Error::e_explain(
                    ErrorType::HttpHeaderLarge,
                    "request header is too large",
                );
            }

            if next_read_timeout {
                return Error::e_explain(
                    ErrorType::ReadTimedout,
                    "read request header timeout",
                );
            }

            let last_buf = self.last_buf.as_mut().unwrap();

            // 写空间不足, 压实一下.
            if last_buf.write_slice_len() < HEADER_BUF_MIN_WRITEABLE_SIZE {
                last_buf.compaction();
            }

            let write_slice = last_buf.write_slice();
            let read_event = self.hc.read(write_slice);
            let read_event = timeout(read_header_timeout, read_event);

            let start_ms = get_now_msec();

            let res = read_event
                .await
                .if_err(
                    ErrorType::ReadTimedout,
                    "read response header timeout",
                )?
                .if_err(ErrorType::ReadError, "read response header error")?;

            let interval = get_now_msec() - start_ms;

            // 低于3ms, 下次读取直接返回Timedout.
            read_header_timeout = read_header_timeout
                .saturating_sub(Duration::from_millis(interval));
            if read_header_timeout.as_millis() < 3 {
                next_read_timeout = true;
            }

            // Read EOF.
            if res == 0 {
                if has_data {
                    return Error::e_explain(
                        ErrorType::ReadEOF,
                        "client closed in read request header process",
                    );
                }
                return Ok(None);
            }

            last_buf.fill(res);
            has_data = true;
        }
    }

    pub fn request(&mut self) -> &mut Request {
        &mut self.req
    }

    pub async fn read_request_body(
        &mut self,
        bufs: &mut VecDeque<Bytes>,
    ) -> Result<HttpResult> {
        match &mut self.body_reader {
            Some(reader) => reader.read_body(&mut self.hc, bufs).await,
            None => Error::e_explain(
                ErrorType::InternalError,
                "body reader no inited",
            ),
        }
    }

    pub async fn read_request_done(&mut self) -> bool {
        match &self.body_reader {
            Some(reader) => reader.done(),
            None => false,
        }
    }

    pub fn expect_trailers_num(&self) -> usize {
        self.expect_trailers
    }

    // 必须在读取完body之后才可以获取req_trailer_headers.
    pub fn get_req_trailer_headers(&self) -> Result<Option<&HttpHeaders>> {
        match &self.body_reader {
            Some(reader) => reader.get_trailers(),
            None => Error::e_explain(
                ErrorType::InternalError,
                "body reader no inited",
            ),
        }
    }

    pub async fn send_resp_header(
        &mut self,
        resp: &mut Response,
    ) -> Result<()> {
        // 我们不响应非101的其他1XX状态码.
        if resp.status_code >= 100
            && resp.status_code <= 199
            && resp.status_code != StatusCode::HTTP_SWITCHING_PROTOCOLS
        {
            debug!("ignoring informational header");
            return Ok(());
        }

        if self.header_send {
            return Error::e_explain(
                ErrorType::InternalError,
                "response header already send done",
            );
        }
        self.header_send = true;

        if self.req.method == Method::HEAD {
            self.resp_no_body = true;
        }

        if resp.status_code < StatusCode::HTTP_STATUS_START
            || resp.status_code > StatusCode::HTTP_STATUS_END
        {
            resp.status_code = StatusCode::HTTP_INTERNAL_SERVER_ERROR;
        }

        self.resp_status_code = resp.status_code;

        if !http_body_allow_status(resp.status_code) {
            self.resp_no_body = true;
        }

        if resp.status_reason.is_empty() {
            resp.status_reason =
                get_status_reason(resp.status_code).to_string();
        }

        let version = self.req.version;

        if PROCESS_EXIT.load(Ordering::Relaxed) {
            self.want_close = true;
        }

        if resp.status_code < 200 || resp.status_code > 399 {
            self.want_close = true;
        }

        if self.want_close {
            resp.headers.set_header(
                Bytes::from_static(HeaderName::CONNECTION),
                Bytes::from_static(HeaderValue::CLOSE),
            );
        }

        // HTTP/1.0 不支持Chunked传输.
        if resp.version == Version::HTTP_10 {
            resp.headers.del_header_all(HeaderName::TRANSFER_ENCODING);
        }

        if let Some(h) = resp.headers.get_header(HeaderName::TRANSFER_ENCODING)
        {
            if h.value_contain(HeaderValue::CHUNKED) {
                self.resp_chunked_transfer = true;
            }
        }

        if self.resp_chunked_transfer {
            if let Some(hh) = resp.headers.get_header(HeaderName::TRAILER) {
                let iter = hh.get_value().split(|c| *c == b',');
                for mut v in iter {
                    v = v.trim_ascii();
                    if !v.is_empty() {
                        self.expect_write_trailers += 1;
                    }
                }
            }
        }

        // 是否为Content-Length Transfer.
        if !self.resp_chunked_transfer {
            if let Some(h) = resp.headers.get_header(HeaderName::CONTENT_LENGTH)
            {
                let header_value = h.get_value();
                let res = String::from_utf8_lossy(header_value)
                    .to_string()
                    .parse()
                    .if_err(
                        ErrorType::InvalidHttpHeader,
                        "invalid content-length header",
                    )?;
                self.resp_content_length = Some(res);
            }
        }

        // HTTP/1.1协议下, 如果没有执行传输方式, 且需要发送body, 默认添加Chunked Transfer.
        if version == Version::HTTP_11
            && !self.resp_chunked_transfer
            && self.resp_content_length.is_none()
            && !self.resp_no_body
        {
            resp.headers.set_header(
                Bytes::from_static(HeaderName::TRANSFER_ENCODING),
                Bytes::from_static(HeaderValue::CHUNKED),
            );
        }

        // 我们不支持HTTP/0.9的处理.
        let version_str;
        if version == Version::HTTP_10 {
            version_str = "HTTP/1.0";
        } else if version == Version::HTTP_11 {
            version_str = "HTTP/1.1";
        } else {
            return Error::e_explain(
                ErrorType::NotImplemented,
                format!("invalid http version: {:#?}", resp.version),
            );
        }

        let status_code = resp.status_code.to_string();

        // response line:
        // HttpVersion Status Reason\r\n
        // eg:
        // HTTP/1.1 200 OK\r\n
        let mut len = version_str.len()
            + 1
            + status_code.len()
            + 1
            + resp.status_reason.len()
            + 2;

        for h in resp.headers.iter() {
            // HeaderName: HeaderValue\r\n
            len += h.get_name().len() + 2 + h.get_value().len() + 2;
        }

        // \r\n
        len += 2;

        let mut buf = BytesMut::with_capacity(len);

        // HttpVersion Status Reason\r\n
        buf.put_slice(version_str.as_bytes());
        buf.put_bytes(b' ', 1);
        buf.put_slice(status_code.as_bytes());
        buf.put_bytes(b' ', 1);
        buf.put_slice(resp.status_reason.as_bytes());
        buf.put_slice(b"\r\n");

        // Http Headers.
        for h in resp.headers.iter() {
            // HeaderName: HeaderValue\r\n
            buf.put_slice(h.get_name());
            buf.put_slice(b": ");
            buf.put_slice(h.get_value());
            buf.put_slice(b"\r\n");
        }

        // Empty line
        buf.put_slice(b"\r\n");

        let bytes = buf.freeze();

        let write_event = self.hc.write_all(&bytes);
        let write_event = timeout(
            Duration::from_millis(self.write_header_timeout_ms as u64),
            write_event,
        );

        write_event
            .await
            .if_err(ErrorType::WriteTimedout, "send response header timeout")?
            .if_err(ErrorType::WriteError, "send response header failed")?;

        self.init_body_writer();

        Ok(())
    }

    pub async fn write_resp_body(
        &mut self,
        bufs: &mut VecDeque<Bytes>,
        done: bool,
    ) -> Result<HttpResult> {
        match &mut self.body_writer {
            Some(writer) => writer.write_body(&mut self.hc, bufs, done).await,

            None => Error::e_explain(
                ErrorType::InternalError,
                "body_writer no inited",
            ),
        }
    }

    pub fn expect_write_trailers_num(&mut self) -> usize {
        self.expect_write_trailers
    }

    pub async fn write_resp_trailers(
        &mut self,
        headers: &HttpHeaders,
    ) -> Result<()> {
        if !self.write_resp_body_done() {
            return Error::e_explain(
                ErrorType::InternalError,
                "response body has not been sent completely",
            );
        }

        if self.write_trailers_done {
            return Error::e_explain(
                ErrorType::InternalError,
                "response trailers already write",
            );
        }

        // 某些传输场景不支持trailers. 此处我们直接返回Ok.
        if self.expect_write_trailers == 0 {
            self.write_trailers_done = true;
            return Ok(());
        }

        self.body_writer
            .as_mut()
            .unwrap()
            .write_trailers(&mut self.hc, headers)
            .await?;
        self.write_trailers_done = true;
        Ok(())
    }

    fn write_resp_body_done(&self) -> bool {
        match &self.body_writer {
            Some(writer) => writer.done(),
            None => false,
        }
    }

    pub fn write_resp_done(&self) -> bool {
        if !self.write_resp_body_done() {
            return false;
        }

        // write_resp_done要额外判断trailers是否发送.
        if self.expect_write_trailers > 0 && !self.write_trailers_done {
            return false;
        }

        true
    }

    pub async fn finalize(&mut self, abort: bool) {
        if self.finalized {
            return;
        }
        self.finalized = true;

        if abort {
            self.want_close = true;
        }

        if !self.write_resp_done() {
            self.want_close = true;
        }

        let read_done = match &self.body_reader {
            Some(reader) => reader.done(),
            None => {
                self.want_close = true;
                false
            }
        };

        // 期望keepalive, 并且request body未读取完毕.
        // discard request body.
        if !self.want_close && !read_done {
            let mut discard_bytes = 0;
            let bufs = &mut VecDeque::with_capacity(4);
            loop {
                if discard_bytes >= self.max_discard_request_body {
                    self.want_close = true;
                    break;
                }

                match self.read_request_body(bufs).await {
                    Ok(res) => {
                        // read success.
                        for v in bufs.iter() {
                            discard_bytes += v.len();
                        }

                        bufs.clear();

                        if res == HttpResult::Complete {
                            break;
                        }
                    }

                    Err(_) => {
                        self.want_close = true;
                        break;
                    }
                }
            }
        }

        // 此时linker虽然已经处于exit状态, 但是未向客户端发送Connection: close, 客户端可能重用该连接.
        // 此时不可以直接关闭连接.
        // if PROCESS_EXIT.load(Ordering::Relaxed) {
        //     self.want_close = true;
        // }

        if self.want_close {
            _ = self.hc.shutdown().await;
            return;
        }

        // for http pipline.
        if let Some(reader) = self.body_reader.take() {
            let bytes = reader.get_remaining_data();
            if !bytes.is_empty() {
                self.hc.set_pre_buf(bytes);
            }
        }
    }

    // 受Rust所有权影响, 直接消费self会导致proxy在finalize之后无法获取到server_session中的一些状态.
    // 所以我们额外引入release接口.
    // 通过finalize可以关闭server_session. 通过release可以释放self中相关的资源.
    pub fn release(self) -> Option<BoxHttpConn> {
        if !self.finalized {
            log::error!("expect server session finalized, but got false");
            return None;
        }

        if self.want_close {
            return None;
        }

        Some(self.hc)
    }
}

impl ServerSession {
    #[inline]
    fn update_session_conf(
        &mut self,
        core_runtime_conf: &Arc<HttpCoreRuntimeConf>,
    ) {
        self.one_header_buf_size = core_runtime_conf.one_header_buf_size;
        self.max_header_size = core_runtime_conf.max_header_size;
        self.read_header_timeout_ms =
            core_runtime_conf.server_read_header_timeout_ms;
        self.write_header_timeout_ms =
            core_runtime_conf.server_write_header_timeout_ms;
        self.body_read_timeout_ms = core_runtime_conf.body_read_timeout_ms;
        self.body_write_timeout_ms = core_runtime_conf.body_write_timeout_ms;
        self.body_buf_size = core_runtime_conf.body_buf_size;
        self.ignore_invalid_header = core_runtime_conf.ignore_invalid_header;
        self.max_discard_request_body =
            core_runtime_conf.max_discard_request_body;
        self.server_keepalive_timeout_ms =
            core_runtime_conf.server_keepalive_timeout_ms;
        self.send_100_continue = core_runtime_conf.send_100_continue;
    }

    // Ok(true): 成功读取到数据.
    // Ok(false): 因超时、ReadEOF等导致读取取失败
    // Err(err): 其他错误情况
    async fn wait_keepalive_request(&mut self) -> Result<bool> {
        let hc = self.hc.as_mut();
        let last_buf = self.last_buf.as_mut().unwrap();

        let unique_id = hc.unique_id();

        let read_event = hc.read(last_buf.write_slice());
        let read_event = timeout(
            Duration::from_millis(self.server_keepalive_timeout_ms as u64),
            read_event,
        );

        let (sender, receiver) = oneshot::channel::<()>();
        ServerConnPool::add_cancel_token(
            unique_id,
            ServerConnCancelToken::new(sender),
        );

        let res = select! {
            _res = read_event => {
                match _res {
                    Ok(r) => r,
                    Err(_) => {
                        // keepalive timeout
                        ServerConnPool::remove_cance_token(unique_id);
                        return Ok(false);
                    }
                }
            },

            _ = receiver => {
                // keepalive abort.
                ServerConnPool::remove_cance_token(unique_id);
                return Ok(false);
            }
        };

        ServerConnPool::remove_cance_token(unique_id);

        match res {
            // Read EOF.
            Ok(0) => Ok(false),
            // Read Success.
            Ok(n) => {
                last_buf.fill(n);
                Ok(true)
            }
            // Read Error.
            Err(e) => Error::e_explain(ErrorType::ReadError, e.to_string()),
        }
    }

    fn parse_request(&mut self) -> Result<HttpResult> {
        loop {
            let line_len = {
                let read_slice = self.last_buf.as_ref().unwrap().read_slice();
                match http_read_crlf_line(read_slice) {
                    Some(n) => n,
                    None => {
                        // Big header line, return error.
                        if read_slice.len()
                            > self.one_header_buf_size
                                - (2 * HEADER_BUF_MIN_WRITEABLE_SIZE)
                        {
                            return Error::e_explain(ErrorType::BigHeaderLine, format!("found big header line, length {} more than {}", read_slice.len(), self.one_header_buf_size - (2*HEADER_BUF_MIN_WRITEABLE_SIZE)));
                        }

                        // No more data.
                        return Ok(HttpResult::Partial);
                    }
                }
            };

            self.parse_request_line(line_len, self.ignore_invalid_header)?;

            let last_buf = self.last_buf.as_mut().unwrap();
            self.header_bytes += line_len;
            last_buf.consume(line_len);

            if self.parse_rh_done {
                return Ok(HttpResult::Complete);
            }
        }
    }

    fn parse_request_line(
        &mut self,
        line_len: usize,
        ignore_invalid_header: bool,
    ) -> Result<()> {
        if !self.parse_rl_done {
            return self.parse_request_line_helper(line_len);
        }

        if !self.parse_rh_done {
            return self.parse_request_header(line_len, ignore_invalid_header);
        }

        unreachable!("request header already parse done");
    }

    fn parse_request_line_helper(&mut self, line_len: usize) -> Result<()> {
        let line_slice =
            &self.last_buf.as_ref().unwrap().read_slice()[0..line_len];

        let mut off: usize = 0;
        let end = line_len;

        let mut method_parse_success = false;
        let method_start = off;

        for (i, v) in line_slice.iter().enumerate().take(end).skip(off) {
            if *v == b' ' {
                let method = Method::from_bytes(&line_slice[method_start..i])
                    .if_err(
                    ErrorType::InvalidRequestLine,
                    "invalid method",
                )?;
                self.req.method = method;
                method_parse_success = true;
                off = i;
                break;
            }

            if !v.is_ascii_uppercase() && *v != b'_' && *v != b'-' {
                return Error::e_explain(
                    ErrorType::InvalidRequestLine,
                    "invalid request method",
                );
            }
        }

        if !method_parse_success {
            return Error::e_explain(
                ErrorType::InvalidRequestLine,
                "invalid request method",
            );
        }

        // parse space before request uri
        for (i, v) in line_slice.iter().enumerate().take(end).skip(off) {
            if *v != b' ' {
                off = i;
                break;
            }
        }

        let uri_start = off;
        for (i, v) in line_slice.iter().enumerate().take(end).skip(off) {
            if *v == b' ' || *v == b'\r' || *v == b'\n' {
                off = i;
                break;
            }
        }

        let uri = Uri::try_from(&line_slice[uri_start..off])
            .if_err(ErrorType::InvalidRequestLine, "invalid request uri")?;
        self.req.uri = uri;

        // parse space before http version
        for (i, v) in line_slice.iter().enumerate().take(end).skip(off) {
            if *v != b' ' {
                off = i;
                break;
            }
        }

        if let Some(host) = self.req.uri.host() {
            let host = host.to_string();
            self.set_session_conf(&host)?;
        }

        // parse request success
        // HTTP Version解析失败当做HTTP_09处理即可, 不再进行错误检查.
        self.parse_rl_done = true;

        let line_slice =
            &self.last_buf.as_ref().unwrap().read_slice()[0..line_len];
        let version = &line_slice[off..];
        if version.len() < 8 {
            self.req.version = Version::HTTP_09;
            return Error::e_explain(
                ErrorType::NotImplemented,
                "no support HTTP/0.9",
            );
        }

        if version[7] == b'0' {
            self.req.version = Version::HTTP_10;
        } else if version[7] == b'1' {
            self.req.version = Version::HTTP_11;
        } else {
            self.req.version = Version::HTTP_09;
        }

        Ok(())
    }

    fn parse_request_header(
        &mut self,
        line_len: usize,
        ignore_invalid_header: bool,
    ) -> Result<()> {
        let line_slice =
            &self.last_buf.as_ref().unwrap().read_slice()[0..line_len];
        let res = parse_http_header(line_slice);
        match res {
            Ok(kv) => {
                if kv.0.is_empty() {
                    self.parse_rh_done = true;
                    return Ok(());
                }

                let header = HttpHeader::with_kv(
                    Bytes::copy_from_slice(kv.0),
                    Bytes::copy_from_slice(kv.1),
                );
                self.process_header(&header)?;
                self.req.headers.append_header(header);

                Ok(())
            }
            Err(e) => {
                if ignore_invalid_header {
                    self.invalid_headers += 1;
                    return Ok(());
                }

                Err(e)
            }
        }
    }

    #[inline]
    fn process_header(&mut self, header: &HttpHeader) -> Result<()> {
        let header_name = header.get_name();

        match header_name.len() {
            4 => {
                if header_name.eq_ignore_ascii_case(HeaderName::HOST) {
                    let host_str =
                        String::from_utf8_lossy(header.get_value()).to_string();

                    self.set_session_conf(&host_str)?;
                    return Ok(());
                }

                Ok(())
            }

            6 => {
                if header_name.eq_ignore_ascii_case(HeaderName::EXPECT)
                    && header.get_value().eq_ignore_ascii_case(b"100-continue")
                {
                    self.expect_100_continue = true;
                    return Ok(());
                }

                Ok(())
            }

            7 => {
                if header_name.eq_ignore_ascii_case(HeaderName::TRAILER) {
                    let iter = header.get_value().split(|c| *c == b',');
                    for mut v in iter {
                        v = v.trim_ascii();
                        if !v.is_empty() {
                            self.expect_trailers += 1;
                        }
                    }
                    return Ok(());
                }

                Ok(())
            }

            10 => {
                if header_name.eq_ignore_ascii_case(HeaderName::CONNECTION) {
                    if header.value_contain(HeaderValue::CLOSE) {
                        self.want_close = true;
                        return Ok(());
                    }

                    if self.req.version == Version::HTTP_10
                        && header.value_contain(HeaderValue::KEEP_ALIVE)
                    {
                        self.want_close = false;
                        return Ok(());
                    }
                }

                Ok(())
            }

            14 => {
                if header_name.eq_ignore_ascii_case(HeaderName::CONTENT_LENGTH)
                {
                    let value_str =
                        String::from_utf8_lossy(header.get_value()).to_string();
                    match value_str.parse() {
                        Ok(cl) => self.content_length = Some(cl),
                        Err(e) => {
                            return Err(Error::from_error(
                                ErrorType::InvalidHttpHeader,
                                "parse Content-Length header failed",
                                e,
                            ));
                        }
                    }
                    return Ok(());
                }

                Ok(())
            }

            17 => {
                if header_name
                    .eq_ignore_ascii_case(HeaderName::TRANSFER_ENCODING)
                {
                    if self.req.version == Version::HTTP_11
                        && header.value_contain(HeaderValue::CHUNKED)
                    {
                        self.chunked_transfer = true;
                    }

                    return Ok(());
                }

                Ok(())
            }

            _ => {
                // Nothing
                Ok(())
            }
        }
    }

    #[inline]
    async fn process_request(&mut self) -> Result<()> {
        if self.expect_100_continue && self.send_100_continue {
            self.hc
                .write_all(b"HTTP/1.1 100 Continue\r\n\r\n")
                .await
                .if_err(ErrorType::WriteError, "while send 100-continue")?;
        }

        self.init_body_reader();

        Ok(())
    }

    #[inline]
    pub fn set_session_conf(&mut self, host: &str) -> Result<()> {
        let listener_address =
            &self.hc.listener.as_ref().unwrap().get_conf().address;
        let address_site_conf =
            HttpSiteRuntimeConfManager::lookup_address_site_conf(
                host,
                listener_address.as_bytes(),
            )?;

        match address_site_conf {
            Some(conf) => {
                self.hc.address_site_conf_info =
                    Some(HttpAddressSiteRuntimeConfInfo {
                        conf: conf.clone(),
                        valid: true,
                    });
                self.update_session_conf(&conf.core_runtime_conf);
            }

            None => {
                let main_conf = get_http_main_runtime_conf();
                self.update_session_conf(&main_conf.core_runtime_conf);
            }
        }

        Ok(())
    }

    fn init_body_reader(&mut self) {
        let last_buf = self.last_buf.take();
        if self.req.is_upgrade_req() {
            self.expect_trailers = 0;
            self.want_close = true;
            self.body_reader = Some(BodyReader::with_until_closed_transfer(
                last_buf,
                self.body_buf_size,
                self.body_read_timeout_ms,
            ));
            return;
        }

        if self.chunked_transfer {
            self.body_reader = Some(BodyReader::with_chunked_transfer(
                last_buf,
                self.body_buf_size,
                self.body_read_timeout_ms,
                self.expect_trailers,
            ));
            return;
        }

        if let Some(cl) = self.content_length {
            self.expect_trailers = 0;
            self.body_reader = Some(BodyReader::with_cl_transfer(
                last_buf,
                cl,
                self.body_buf_size,
                self.body_read_timeout_ms,
            ));
            return;
        }

        self.expect_trailers = 0;
        self.body_reader = Some(BodyReader::with_no_body(last_buf));
    }

    fn init_body_writer(&mut self) {
        if self.resp_no_body {
            self.want_close = true;
            self.expect_write_trailers = 0;
            self.body_writer = Some(BodyWriter::with_no_body());
            return;
        }

        if self.resp_status_code == StatusCode::HTTP_SWITCHING_PROTOCOLS {
            self.want_close = true;
            self.expect_write_trailers = 0;
            self.body_writer = Some(BodyWriter::with_until_closed_transfer(
                self.body_write_timeout_ms,
            ));
            return;
        }

        if self.resp_chunked_transfer {
            self.body_writer = Some(BodyWriter::with_chunked_transfer(
                self.body_write_timeout_ms,
                self.expect_write_trailers,
            ));
            return;
        }

        if let Some(cl) = self.resp_content_length {
            self.expect_write_trailers = 0;
            self.body_writer = Some(BodyWriter::with_cl_transfer(
                cl,
                self.body_write_timeout_ms,
            ));
            return;
        }

        self.want_close = true;
        self.expect_write_trailers = 0;
        self.body_writer = Some(BodyWriter::with_until_closed_transfer(
            self.body_write_timeout_ms,
        ));
    }
}

#[cfg(test)]
mod tests {
    use http::{uri::Scheme, Uri};

    #[test]
    fn test_parse_request_line() {
        let str1 = "http://www.aaa.com:1212/1.txt?a=1&b=2";
        let uri = Uri::try_from(str1).unwrap();
        assert_eq!(*uri.scheme().unwrap(), Scheme::HTTP);
        assert_eq!(uri.authority().unwrap().as_str(), "www.aaa.com:1212");
        assert_eq!(uri.host().unwrap(), "www.aaa.com");
        assert_eq!(uri.port().unwrap().as_str(), "1212");
        assert_eq!(uri.path(), "/1.txt");
        assert_eq!(uri.path_and_query().unwrap().as_str(), "/1.txt?a=1&b=2");
        assert_eq!(uri.query().unwrap(), "a=1&b=2");

        // 判断是否支持CONNECT格式的request line.
        let str2 = "www.abc.com:443";
        let uri = Uri::try_from(str2).unwrap();
        assert_eq!(uri.authority().unwrap().as_str(), "www.abc.com:443");
        assert!(uri.path_and_query().is_none());
    }
}
